package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Date 2021/7/29 14:54
 * 有订单数据,格式为: (订单ID，用户ID，时间戳/事件时间，订单金额)
 * 要求每隔5s,计算5秒内，每个用户的订单总金额
 * 并添加Watermark来解决一定程度上的数据延迟和数据乱序问题。
 * 并使用OutputTag+allowedLateness解决3s延迟数据丢失问题
 */
public class AllowedLatenessDemo {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.Source 创建 Order 类 orderId:String userId:Integer money:Integer eventTime:Long
        //模拟生成 Order 数据  事件时间=当前时间-5秒钟随机*1000
        DataStreamSource<Orders> source = env.addSource(new CustomOrder());
        //3.Transformation
        //-告诉Flink要基于事件时间来计算!
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//新版本默认就是EventTime
        //-分配水印机制，最多延迟3秒，告诉Flink数据中的哪一列是事件时间,因为Watermark = 当前最大的事件时间 - 最大允许的延迟时间或乱序时间
        //定义一个严重迟到的侧输出流
        OutputTag serious_lateness_data = new OutputTag("serious_lateness_data", TypeInformation.of(Orders.class));

        SingleOutputStreamOperator<Orders> result = source.assignTimestampsAndWatermarks(WatermarkStrategy

                .<Orders>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((element, recordTimestamp) -> element.eventTime))
                //代码走到这里,就已经被添加上Watermark了!接下来就可以进行窗口计算了
                //要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
                .keyBy(t -> t.uid)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .allowedLateness(Time.seconds(3))
                .sideOutputLateData(serious_lateness_data)
                .sum("money");
        //4.Sink
        result.print("正常数据:");
        result.getSideOutput(serious_lateness_data).print("严重迟到的数据:");
        //5.execute
        env.execute();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Orders{
        private String oid;
        private int uid;
        private int money;
        private long eventTime;
    }

    private static class CustomOrder extends RichSourceFunction<Orders> {
        //定义表级
        boolean flag = true;
        //定义一个随机变量
        Random rn = new Random();
        @Override
        public void run(SourceContext<Orders> ctx) throws Exception {
            while(flag){
                String oid = UUID.randomUUID().toString();
                int uid = rn.nextInt(3);
                int money = rn.nextInt(101);
                long eventTime = System.currentTimeMillis()-rn.nextInt(10000);
                ctx.collect(new Orders(oid,uid,money,eventTime));
                Thread.sleep(200);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
